package com.sloan.community.ws;

import com.sloan.community.common.dto.NoticeDTO;
import com.sloan.community.common.res.Result;
import com.sloan.community.exception.CommunityException;
import com.sloan.community.util.JsonUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.stereotype.Component;
import org.springframework.util.AntPathMatcher;
import org.springframework.web.reactive.socket.WebSocketHandler;
import org.springframework.web.reactive.socket.WebSocketMessage;
import org.springframework.web.reactive.socket.WebSocketSession;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;

import static com.sloan.community.common.Constant.NoticeUrlPattern.PATTERN;
import static com.sloan.community.common.Constant.NoticeUrlPattern.USER_ID;

/**
 * @Author: yaosj
 * @CreateTime: 2024-12-23
 */
@Slf4j
@Component
public class CommunityWebSocketHandler implements WebSocketHandler {

    private static final AntPathMatcher PATH_MATCHER = new AntPathMatcher();

    private static final Map<Long, WebSocketSession> SESSIONS = new ConcurrentHashMap<>(256);

    @Override
    public Mono<Void> handle(WebSocketSession session) {
        String path = session.getHandshakeInfo().getUri().getPath();
        boolean match = PATH_MATCHER.match(PATTERN, path);
        if (!match) {
            return Mono.error(CommunityException.of("链接错误"));
        }

        Long userId = Long.parseLong(PATH_MATCHER.extractUriTemplateVariables(PATTERN, path).get(USER_ID));
        SESSIONS.putIfAbsent(userId, session);

        return session.receive()
                .flatMap(msg -> {
                    String payloadAsText = msg.getPayloadAsText();
                    NoticeDTO noticeDTO = JsonUtils.deserialize(payloadAsText, NoticeDTO.class);
                    Long toUserId = noticeDTO.getToUserId();
                    // 如果目标用户在线，转发消息
                    if (SESSIONS.containsKey(toUserId)) {
                        WebSocketSession targetSession = SESSIONS.get(toUserId);
                        if (targetSession.isOpen()) {
                            WebSocketMessage socketMessage = session.textMessage(
                                    JsonUtils.serialize(Result.success(noticeDTO.copyUpdateMsg("+1"))));
                            return targetSession.send(Mono.just(socketMessage));
                        } else {
                            SESSIONS.remove(toUserId);
                        }

                    }

                    // 当前用户不在线
                    return session.send(Mono.just(session.textMessage(JsonUtils.serialize(Result.onLine()))));
                })
                .doOnError(err -> {
                    log.error("建立连接发送消息过程出现错误:{}", err.getMessage());
                })
                .then(Mono.defer(() -> {
                    log.info("用户: {} 链接close", userId);
                    return sessionClose(userId);
                }))
                .doFinally(signalType -> SESSIONS.remove(userId));

    }

    public Mono<Void> pushMsg(NoticeDTO noticeDTO) {
        Long toUserId = noticeDTO.getToUserId();
        if (Objects.isNull(toUserId)) {
            return Mono.error(CommunityException.of("建立长连接发送消息的时候,发送人ID不能为null"));
        }

        if (SESSIONS.containsKey(toUserId)) {
            WebSocketSession socketSession = SESSIONS.get(toUserId);
            if (socketSession.isOpen()) {
                WebSocketMessage textMessage = socketSession.textMessage(JsonUtils.serialize(Result.success(noticeDTO.copyUpdateMsg("+1"))));
                return socketSession.send(Mono.just(textMessage));
            }
        }
        return Mono.empty();
    }

    public Mono<Void> sessionClose(Long userId) {
        if (SESSIONS.containsKey(userId)) {
            return SESSIONS.get(userId).close();
        }
        return Mono.empty();
    }
}
